// 核心思想：借助 shared_ptr 实现 Copy-On-Write
// 共享只读快照:
// - 连接列表由 shared_ptr<ConnectionList> 持有。读取路径只需获取 shared_ptr 的副本即可增加引用计数，在锁外遍历集合，不与其他读者竞争。
// 
// 写前检测独占权:
// - 写操作（新增/删除连接）前检查 shared_ptr::unique()。若不是唯一持有者，说明仍有读者引用旧集合，需要拷贝一份副本并在副本上修改，从而保证读者看到的是稳定快照。
// 
// 读多写少场景优化:
// - 在广播消息时，读操作频繁且无锁；写操作偶尔触发时才复制数据，利用 Copy-On-Write 将锁粒度降到最低，提高多线程聊天室广播的整体效率。

#include "codec.h"

#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include <set>
#include <stdio.h>
#include <unistd.h>

using namespace muduo;
using namespace muduo::net;

// 借助 shared_ptr 实现“写时复制”（Copy-On-Write, COW）的聊天室服务端。
// 关键点：
// 1. connections_ 是一个指向 ConnectionList 的 shared_ptr，所有需要“只读”访问连接列表的地方
//    都能共享同一份数据，避免频繁加锁。
// 2. 当有写操作（新增/删除连接）时，只有当 shared_ptr 不是唯一持有者时才复制一份副本，
//    然后在副本上修改，确保读者看到的是旧的稳定快照。
class ChatServer : noncopyable
{
 public:
  ChatServer(EventLoop* loop,
             const InetAddress& listenAddr)
  : server_(loop, listenAddr, "ChatServer"),
    // codec_ 负责长度前缀编解码，onStringMessage 是业务回调。
    codec_(std::bind(&ChatServer::onStringMessage, this, _1, _2, _3)),
    // 初始 connections_ 指向一个空的 ConnectionList
    connections_(new ConnectionList)
  {
    // 注册连接回调与消息回调
    server_.setConnectionCallback(
        std::bind(&ChatServer::onConnection, this, _1));
    server_.setMessageCallback(
        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
  }

  void setThreadNum(int numThreads)
  {
    server_.setThreadNum(numThreads);
  }

  void start()
  {
    server_.start();
  }

 private:
  // onConnection 在对应的 I/O 线程中执行，用于维护连接集合
  void onConnection(const TcpConnectionPtr& conn)
  {
    LOG_INFO << conn->peerAddress().toIpPort() << " -> "
        << conn->localAddress().toIpPort() << " is "
        << (conn->connected() ? "UP" : "DOWN");

    MutexLockGuard lock(mutex_);
    // ---- Copy-On-Write 核心 ----
    // 如果当前 connections_ 被多个 shared_ptr 持有（说明有读者在使用旧集合的快照），
    // 就复制一份新的 ConnectionList 作为本次修改的基础，这样读者仍旧使用旧数据，互不干扰。
    if (!connections_.unique())
    {
      connections_.reset(new ConnectionList(*connections_));
    }
    // 断言此时 connections_ 是唯一持有者，可以安全修改。
    assert(connections_.unique());

    if (conn->connected())
    {
      // 写操作发生在独占副本上
      connections_->insert(conn);
    }
    else
    {
      connections_->erase(conn);
    }
  }

  typedef std::set<TcpConnectionPtr> ConnectionList;
  typedef std::shared_ptr<ConnectionList> ConnectionListPtr;

  // onStringMessage 只需要“读”连接集合，因此可直接拿到 shared_ptr 快照后无锁遍历
  void onStringMessage(const TcpConnectionPtr&,
                       const string& message,
                       Timestamp)
  {
    // getConnectionList 返回一个共享快照；因为 shared_ptr 计数增加，
    // onConnection 如需写入时会触发 COW。
    ConnectionListPtr connections = getConnectionList();
    for (ConnectionList::iterator it = connections->begin();
        it != connections->end();
        ++it)
    {
      codec_.send(get_pointer(*it), message);
    }
  }

  ConnectionListPtr getConnectionList()
  {
    MutexLockGuard lock(mutex_);
    // 返回 shared_ptr 的拷贝，增加引用计数。
    // 读者拿到快照后可以在锁外遍历，期间不会受到写操作影响。
    return connections_;
  }

  TcpServer server_;
  LengthHeaderCodec codec_;
  MutexLock mutex_;
  // connections_ 是受 mutex_ 保护的 shared_ptr，负责托管连接集合。
  // 使用 COW 策略，实现读多写少场景下的高效广播。
  ConnectionListPtr connections_ GUARDED_BY(mutex_);
};

int main(int argc, char* argv[])
{
  LOG_INFO << "pid = " << getpid();
  if (argc > 1)
  {
    EventLoop loop;
    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));
    InetAddress serverAddr(port);
    ChatServer server(&loop, serverAddr);
    if (argc > 2)
    {
      server.setThreadNum(atoi(argv[2]));
    }
    server.start();
    loop.loop();
  }
  else
  {
    printf("Usage: %s port [thread_num]\n", argv[0]);
  }
}